package com.zhang.first.day04;

import com.zhang.first.utils.SensorReading;
import com.zhang.first.utils.SensorSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @title: 连续1s传感器温度上升报警
 * @author: zhang
 * @date: 2022/1/16 22:25
 */
public class Example1 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new SensorSource())
                .keyBy(r -> r.sensorId)
                .process(new Alter())
                .print();


        env.execute();
    }

    public static class Alter extends KeyedProcessFunction<String, SensorReading, String> {
        private ValueState<Integer> lastTemp;
        private ValueState<Long> timerTs;

        @Override
        public void open(Configuration parameters) throws Exception {
            lastTemp = getRuntimeContext().getState(
                    new ValueStateDescriptor<Integer>("last_temp", Types.INT)
            );
            timerTs = getRuntimeContext().getState(
                    new ValueStateDescriptor<Long>("timer", Types.LONG)
            );
        }

        @Override
        public void processElement(SensorReading value, KeyedProcessFunction<String, SensorReading, String>.Context ctx, Collector<String> out) throws Exception {
            Integer preTemp = null;
            if (lastTemp.value() != null) {
                //来的不是第一条
                //将lastTemp值缓存到preTemp
                preTemp = lastTemp.value();
            }
            //将当前温度保存到lastTemp中
            lastTemp.update(value.temperature);

            Long ts = null;
            if (timerTs.value() != null) {
                //时间戳状态存在，说明定时器存在
                ts = timerTs.value();
            }

            //保证来的数据不是第一条数据
            if (preTemp != null) {
                if (value.temperature > preTemp && ts == null) {
                    //如果温度上升，且报警定时器不存在
                    long ontSecondLater = ctx.timerService().currentProcessingTime() + 1000L;
                    //1 注册定时器
                    ctx.timerService().registerProcessingTimeTimer(ontSecondLater);
                    //2 将保存定时器的时间戳保存到状态变量
                    timerTs.update(ontSecondLater);
                } else if (value.temperature < preTemp && ts != null) {
                    //如果温度出现下降，且报警定时器存在
                    //1.将报警器删除
                    ctx.timerService().deleteProcessingTimeTimer(ts);
                    //2.将保存定时器时间戳的状态变量清空
                    timerTs.clear();
                }
            }
        }

        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<String, SensorReading, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            out.collect("传感器" + ctx.getCurrentKey() + "连续1s温度上升");
            timerTs.clear();
        }
    }
}
